package cn.itcast.streaming.source;

import cn.itcast.streaming.entity.ElectricFenceResultTmp;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * Author itcast
 * Date 2021/9/26 11:24
 * Desc 读取mysql中电子围栏的配置数据
 */
public class MySQLElectricFenceSouce extends RichSourceFunction<HashMap<String, ElectricFenceResultTmp>> {

    private static final Logger logger = LoggerFactory.getLogger(MySQLElectricFenceSouce.class.getSimpleName());
    private static Connection conn = null;
    private static Statement statement = null;
    //设置标识用于记录当前循环读取mysql配置
    private static Boolean flag = true;
    private static String elerulesTime = null;

    //1.重写 open 方法
    @Override
    public void open(Configuration parameters) throws Exception {
        //1.1 获取上下文中的 parameterTool
        ParameterTool globalJobParameters = (ParameterTool) getRuntimeContext()
                .getExecutionConfig().getGlobalJobParameters();
        //1.2 读取配置文件中，注册驱动 url user password
        String driver = globalJobParameters.getRequired("jdbc.driver");
        String url = globalJobParameters.getRequired("jdbc.url");
        String user = globalJobParameters.getRequired("jdbc.user");
        String password = globalJobParameters.getRequired("jdbc.password");
        //读取出来多长时间获取一次mysql的配置数据
        elerulesTime = globalJobParameters.getRequired("elerules.millionseconds");
        //设置驱动和连接
        Class.forName(driver);
        conn = DriverManager.getConnection(url, user, password);
        //1.3 实例化statement
        statement = conn.createStatement();
    }

    //3.重写 run 方法
    @Override
    public void run(SourceContext<HashMap<String, ElectricFenceResultTmp>> ctx) throws Exception {
        while(flag) {
            HashMap<String, ElectricFenceResultTmp> map = new HashMap<>();
            //3.1 每指定时间循环读取 mysql 中的电子围栏规则
            String sql = "select vins.vin,setting.id,setting.name,setting.address,setting.radius,setting.longitude,setting.latitude,setting.start_time,setting.end_time \n" +
                    "from vehicle_networking.electronic_fence_setting setting \n" +
                    "inner join vehicle_networking.electronic_fence_vins vins on setting.id=vins.setting_id \n" +
                    "where setting.status=1";
            ResultSet rs = statement.executeQuery(sql);
            while(rs.next()){
                map.put(
                        rs.getString("vin"),
                        new ElectricFenceResultTmp(
                             rs.getInt("id"),
                                rs.getString("name"),
                                rs.getString("address"),
                                rs.getFloat("radius"),
                                rs.getDouble("longitude"),
                                rs.getDouble("latitude"),
                                rs.getDate("start_time"),
                                rs.getDate("end_time")
                        )
                );
            }
            //每行都收集完发送给下游 task 任务
            ctx.collect(map);
            //关闭rs
            if(!rs.isClosed()) rs.close();
            //3.2 收集 electricFenceResult 指定休眠时间
            TimeUnit.MILLISECONDS.sleep(Long.parseLong(elerulesTime));
        }
    }

    //4.重写 cancel 方法
    @Override
    public void cancel() {
        flag = false;
    }

    //2.重写 close 方法
    @Override
    public void close() throws Exception {
        super.close();
        //2.1 关闭 statement 和 conn
        if(!statement.isClosed()) statement.close();
        if(!conn.isClosed()) conn.close();
    }
}
